package com.lzb.interview;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;
import org.springframework.util.StopWatch;

import com.lzb.common.util.thread.Threads;

/**
 * 获取出现次数最多的字符串集合
 * @author lizebin
 *
 */
public class FindMaxAppearanceOfN {
	
	private final static int SIZE = 3500000;

	private final File file = new File("phone_number_" + String.valueOf(SIZE) + ".txt");
	
	private final static int THREAD_SIZE = 6;
	
	private final static int SELECTED_SIZE = 10;
	
	private final ExecutorService executor = Executors.newFixedThreadPool(THREAD_SIZE);
	
	private static CountDownLatch latch = new CountDownLatch(THREAD_SIZE);
	
	//存储从文件读取的电话号码，支持并发
	private static ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();
	//Map变量一定要共享，因为如果是每个线程各自排序，再合并排序，会有问题
	//例如，获取出现次数最多的前两个：
	//a.线程A-{1:3,2:3},其实后面有{3,3},这里只取前两个，就漏了3
	//b.线程B-{3:1,1:1}
	//如果是线程各自排序再合并排序，最终结果：{1:4, 2:3}；但实际应该是{3:4,2:4}或者{2:4,3:4}
	private static ConcurrentHashMap<String, AtomicInteger> phoneNumCounts = new ConcurrentHashMap<String, AtomicInteger>(SIZE);
	
	private static final Comparator<Map.Entry<String, AtomicInteger>> comparator = new Comparator<Map.Entry<String, AtomicInteger>>() {
		public int compare(Entry<String, AtomicInteger> entry1,
				Entry<String, AtomicInteger> entry2) {
			int value1 = 0, value2 = 0;
			try {
				value1 = entry1.getValue().intValue();
				value2 = entry2.getValue().intValue();
			} catch (NumberFormatException e) {
				value1 = 0;
				value2 = 0;
			}
			return value2 - value1;
		}
	};
	
	//@Test
	public void createFile() throws IOException {
		String[] phoneNumberPreffix = new String[]{"131", "132", "133", "134", "135", "136", "137", "138", "139"};
		Random random = new Random();
		/*
		 * 构造10万个电话号码，存储到文件中
		 * */
		FileOutputStream fos = null;
		PrintWriter pw = null;
		try {
			fos = new FileOutputStream(file);
			pw = new PrintWriter(fos);
			for(int i=0; i<SIZE; i++) {
				pw.println(phoneNumberPreffix[random.nextInt(phoneNumberPreffix.length)] + String.valueOf(10000000 + random.nextInt(89999999)));
			}
		} finally {
			if(pw != null) {
				pw.flush();
				pw.close();
				pw = null;
			}
			if(fos != null) {
				fos.close();
				fos = null;
			}
		}
	}
	
	@Test
	public void execute() throws InterruptedException, ExecutionException {
		/*
		 * 1. 启动1个线程用于IO读取电话号码，并放置到队列
		 * 2. 启动1个线程用于处理队列里的电话号码
		 * 3. 最后用一条线程汇总数据
		 * */
		StopWatch time = new StopWatch();
		time.start();
		executor.execute(new Reader(file));
		for(int i=0; i<THREAD_SIZE - 1; i++) {
			executor.execute(new Finder());
		}
		latch.await();
		System.out.println(sortMapByValue(phoneNumCounts, 0, SELECTED_SIZE - 1, comparator));
		time.stop();
		System.out.println("done--------" + time.getTotalTimeSeconds());
		Threads.gracefulShutdown(executor, 60000);
	}
	
	/**
	 * 查找电话号码并排序
	 * @author lizebin
	 *
	 */
	class Finder implements Runnable {
		
		@Override
		public void run() {
			while(Reader.running) {
				String phoneNumber = queue.poll();
				//如果存在则初始化个数为零，反之则次数自增
				if(phoneNumber != null) {
					if(phoneNumCounts.putIfAbsent(phoneNumber, new AtomicInteger(1)) != null) {
						phoneNumCounts.get(phoneNumber).addAndGet(1);
					} 
				}
			} 
			latch.countDown();
		}
	}
	
	/**
	 * 读取文件
	 * @author lizebin
	 *
	 */
	static class Reader implements Runnable {
		public static volatile boolean running = true;
		
		private File file;
		
		public Reader(File file) {
			this.file = file;
		}
		
		@Override
		public void run() {
			FileInputStream fis = null;
			InputStreamReader isr = null;
			BufferedReader br = null;
			try {
				fis = new FileInputStream(file);
				isr = new InputStreamReader(fis);
				br = new BufferedReader(isr);
				String temp = null;
				//读取文件完成
				while((temp = br.readLine()) != null) {
					queue.add(temp);
				}
				//并且队列未消费完，读线程自旋
				while(queue.size() != 0) {}
			} catch(Exception e) {
				e.printStackTrace();
			} finally {
				running = false;
				latch.countDown();
				try {
					if(fis != null) {
						fis.close();
						fis = null;
					}
					if(isr != null) {
						isr.close();
						isr =null;
					}
					if(br != null) {
						br.close();
						br =null;
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	/**
	 * 根据Map的Value进行排序
	 * @param oriMap
	 * @param startIndex
	 * @param endIndex
	 * @param comparator
	 * @return
	 */
	public static Map<String, AtomicInteger> sortMapByValue(Map<String, AtomicInteger> oriMap, int startIndex, int endIndex, Comparator<Map.Entry<String, AtomicInteger>> comparator) {
		int counts = endIndex - startIndex + 1;
		Map<String, AtomicInteger> sortedMap = new LinkedHashMap<String, AtomicInteger>(counts);
		if (oriMap != null && !oriMap.isEmpty()) {
			List<Map.Entry<String, AtomicInteger>> entryList = new LinkedList<Map.Entry<String, AtomicInteger>>(oriMap.entrySet());
			Collections.sort(entryList, comparator);
			if(entryList.size() > counts) {
				entryList = new LinkedList<Map.Entry<String, AtomicInteger>>(entryList.subList(startIndex, endIndex));
			}
			Iterator<Map.Entry<String, AtomicInteger>> iter =  entryList.iterator();
			Map.Entry<String, AtomicInteger> tmpEntry = null;
			while (iter.hasNext()) {
				tmpEntry = iter.next();
				sortedMap.put(tmpEntry.getKey(), tmpEntry.getValue());
			}
		}
		return sortedMap;
	}
}

